package cn.tang.tframe.mq;

/**
 * Created by tanghc on 17/2/3.
 */

import cn.tang.tframe.mq.consumer.IQtMsgProcessor;
import cn.tang.tframe.mq.consumer.QtConsumerClient;
import cn.tang.tframe.test.common.TestUtil;
import cn.tang.tframe.threadpool.QtThreadPoolManager;

import java.io.Serializable;
import java.util.Properties;

public class QtKafkaConsumerExample {
	public static void main(String[] args) throws InterruptedException {
		QtThreadPoolManager.getInstance().addThread(QtConsumerClient.builderConsumer("qtuc", "basedata", "t1", new IQtMsgProcessor() {
			@Override
			public <T extends Serializable> void onConsume(String msgId, T o) {
				System.out.println("1=============:"+msgId+":"+o);
			}
		}));
		QtThreadPoolManager.getInstance().addThread(QtConsumerClient.builderConsumer("qtuc", "basedata", "t1", new IQtMsgProcessor() {
			@Override
			public <T extends Serializable> void onConsume(String msgId, T o) {
				System.out.println("2=============:"+msgId+":"+o);
			}
		}));
		Properties properties = new Properties();
		properties.setProperty("group.id","test1");
		QtThreadPoolManager.getInstance().addThread(QtConsumerClient.builderConsumer("qtuc", "basedata", "t1", properties, new IQtMsgProcessor() {
			@Override
			public <T extends Serializable> void onConsume(String msgId, T o) {
				System.out.println("3=============:" + msgId + ":" + o);
			}
		}));
		TestUtil.waitForInput();
	}

}